package com.atliuzu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atliuzu.app.func.DimAsyncFunction;
import com.atliuzu.bean.TestPaperBean;
import com.atliuzu.utils.ClickHouseUtil;
import com.atliuzu.utils.DateFormatUtil;
import com.atliuzu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class DwsTestPaperWindow {

    public static void main(String[] args) throws Exception {

        //todo 1.获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //todo 2.读取DWD层试卷考试主题 dwd_test_exam 数据创建为流并转换数据结构为JavaBean
        String topic = "dwd_test_exam";
        String groupId = "test_exam_06";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        //todo 3.转换为json对象
        SingleOutputStreamOperator<TestPaperBean> testpaperDS = kafkaDS.flatMap(new FlatMapFunction<String, TestPaperBean>() {
            @Override
            public void flatMap(String s, Collector<TestPaperBean> collector) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(s);
                    collector.collect(TestPaperBean.builder()
                            .paper_id(jsonObject.getString("paper_id"))
                            .total_user(1L)
                            .total_score(jsonObject.getDouble("score"))
                            .total_duration_sec(jsonObject.getLong("duration_sec"))
                            .ts(DateFormatUtil.toTs(jsonObject.getString("create_time"), true))
                            .build());
                } catch (Exception ignored) {
                }
            }
        });

//        testpaperDS.print("test>>>>>>");

        //todo 5.分组 开窗 聚合
        SingleOutputStreamOperator<TestPaperBean> reduceDS = testpaperDS.assignTimestampsAndWatermarks(WatermarkStrategy.<TestPaperBean>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<TestPaperBean>() {
            @Override
            public long extractTimestamp(TestPaperBean testPaperBean, long l) {
                return testPaperBean.getTs();
            }
        }))
                .keyBy(TestPaperBean::getPaper_id)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<TestPaperBean>() {
                    @Override
                    public TestPaperBean reduce(TestPaperBean value1, TestPaperBean value2) throws Exception {
                        value1.setTotal_user(value1.getTotal_user() + value2.getTotal_user());
                        value1.setTotal_score(value1.getTotal_score() + value2.getTotal_score());
                        value1.setTotal_duration_sec(value1.getTotal_duration_sec() + value2.getTotal_duration_sec());
                        return value1;
                    }
                }, new WindowFunction<TestPaperBean, TestPaperBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TestPaperBean> input, Collector<TestPaperBean> out) throws Exception {
                        TestPaperBean next = input.iterator().next();

                        next.setTs(System.currentTimeMillis());
                        next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        next.setStt(DateFormatUtil.toYmdHms(window.getStart()));

                        out.collect(next);
                    }
                });
        //打印测试
        reduceDS.print("reduceDS>>>>>>>>>>");

        //todo 6.关联维表信息
        SingleOutputStreamOperator<TestPaperBean> testPaperWithTestParperDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<TestPaperBean>("DIM_TEST_PAPER") {
                    @Override
                    public String getKey(TestPaperBean input) {
                        return input.getPaper_id();
                    }

                    @Override
                    public void join(TestPaperBean input, JSONObject dimInfo) {
                        input.setPaper_name(dimInfo.getString("PAPER_TITLE"));
                    }
                },
                100, TimeUnit.SECONDS);

        testPaperWithTestParperDS.print(">>>>>>>>>>>");

        //todo 7.将数据写出到ClickHouse
        testPaperWithTestParperDS.addSink(ClickHouseUtil.getJdbcSink("insert into dws_test_paper_window values(?,?,?,?,?,?,?,?)"));
        //todo 8.启动任务
        env.execute("DwsTradeProvinceOrderWindow");
    }
}
